package part03;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;

public class WatermarkDemo {
    public static void main(String[] args) throws Exception {
        //1、获取数据源
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getConfig().setAutoWatermarkInterval(1000L);
        env.setParallelism(1);
        DataStreamSource<String> data = env.addSource(new SourceFunction<String>() {
            int num = 1;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (true) {
                    ctx.collect(num + "," + System.currentTimeMillis());
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });
        //2、转化
        SingleOutputStreamOperator<Tuple2<String, Long>> map = data.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split(",");
                return new Tuple2<>(split[0], Long.valueOf(split[1]));
            }
        });
        //3、声明水印（watermark）
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd:HH-mm-ss.SSS");
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = map.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Tuple2<String, Long>>() {
                    private long maxTime = 0L;

                    @Override
                    public void onEvent(Tuple2<String, Long> stringLongTuple2, long l, WatermarkOutput watermarkOutput) {
                        // 获取最新的事件时间戳
                        maxTime = Math.max(maxTime, stringLongTuple2.f1);
                        System.out.println("maxTimeStamp: " + maxTime + "...format: " + sdf.format(maxTime));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
                        //最大的允许的延迟时间
                        long delayTime = 3000L;
                        //输出触发计算的水印时间
                        watermarkOutput.emitWatermark(new Watermark(maxTime - delayTime));
                        System.out.println("---onPeriodicEmit---");
                    }
                };
            }
        }.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
            @Override
            public long extractTimestamp(Tuple2<String, Long> stringLongTuple2, long l) {
                //返回事件的时间戳
                return stringLongTuple2.f1;
            }
        }));
        //4、分组聚合，调用window的操作
        KeyedStream<Tuple2<String, Long>, String> keyBy = watermarks.keyBy(f -> f.f0);
        WindowedStream<Tuple2<String, Long>, String, TimeWindow> window = keyBy.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<String> apply = window.apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                System.out.println("窗口开始时间：" + sdf.format(window.getStart()));
                ArrayList<Long> list = new ArrayList<>();
                for (Tuple2<String, Long> tuple2 : input) {
                    list.add(tuple2.f1);
                }
                Collections.sort(list);
                String info = "当前数据：" + s
                        + "--" + "窗口数据量:" + list.size()
                        + "--" + "事件开始时间：" + sdf.format(list.get(0))
                        + "--" + "事件结束时间：" + sdf.format(list.get(list.size() - 1))
                        + "--" + "窗口开始时间：" + sdf.format(window.getStart())
                        + "--" + "窗口结束时间：" + sdf.format(window.getEnd());
                out.collect(info);
            }
        });
        //5、保存处理结果
        apply.print();
        env.execute();
    }
}
